//
//  AppSourceThread.cpp
//  2006 John Wiggins
//  Based on code from Jeepers Elvis, 2000 John Wiggins
//
// This program is free software; you can redistribute it and/or modify  
// it under the terms of the GNU General Public License as published by  
// the Free Software Foundation; either version 2 of the License, or     
// ( at your option ) any later version.                                   
// 
// This program is distributed in the hope that it will be useful,       
// but WITHOUT ANY WARRANTY; without even the implied warranty of        
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the         
// GNU General Public License for more details.                          
//     
//     You should have received a copy of the GNU General Public License     
//     along with this program; if not, write to the                         
//         Free Software Foundation, Inc.,                                       
//         59 Temple Place - Suite 330, Boston, MA  02111-1307, USA.  
//
//-------------------------------------------------------------
#include <stdio.h>
#include <stdlib.h>
#include <unistd.h>
#include <sys/time.h>
#include <pthread.h>
#include <assert.h>

#include "App/AppSourceThread.h"
#include "App/AppMessageConstants.h"
//-------------------------------------------------------------
using std::list;
using std::vector;
//-------------------------------------------------------------
AppSourceThread::AppSourceThread()
{
    char thread_name[64];
    
    // spawn our work thread
    if ( 0 != pthread_create( &mThread, NULL, start_loop, ( void * )this ) )
    {
        printf( "error creating the source thread\n" );
    }

    // create our message port
    // the name has to be unique
    snprintf( thread_name, sizeof( thread_name ), "source %d", ( int )mThread );
    mMsgPort = pp_create_port( thread_name );
    // check for error
    if ( mMsgPort < GE_NO_ERROR )
    {
        printf( "Error creating the source thread's port. ( %d )\n", mMsgPort );
        exit( -1 );
    }
    
    // init the surface pool
    // with empty images
    //std::vector<RenSurface *> mSurfacePool;
    mSurfacePool.push_back( new RenSurface( GE_FMT_GRAY, 0, 0 ) );
    mSurfacePool.push_back( new RenSurface( GE_FMT_GRAY, 0, 0 ) );
    mSurfacePool.push_back( new RenSurface( GE_FMT_GRAY, 0, 0 ) );

    // "pointers" to the surfaces
    mOutputSurface = 0;
    mWorkSurface = 1;
    mScratchSurface = 2;

    // we aren't qutting right now
    mQuitting = false;

    // we don't have any plugins yet
    mSrcPlugin = NULL;
    mOutPlugin = NULL;
}
//-------------------------------------------------------------
AppSourceThread::~AppSourceThread()
{
    // send a quit message to the thread
    IpcMessage quitmsg( GE_QUIT );
    quitmsg.SendMessage( mMsgPort );

    // wait for the thread to exit
    pthread_join( mThread, NULL );

    // delete the message port
    pp_delete_port( mMsgPort );

    // delete the surfaces in the pool
    delete mSurfacePool[0];
    delete mSurfacePool[1];
    delete mSurfacePool[2];

    // forget about the plugins, the plugin manager will get rid of them
    mSrcPlugin = NULL;
}
//-------------------------------------------------------------
RenSurface *
AppSourceThread::getSurface()
{
    // XXX: should the surface be locked before returning a pointer to it?
    // ( probably, yeah... how would we know? )
    return mSurfacePool[mOutputSurface];
}
//-------------------------------------------------------------
void
AppSourceThread::receiveMessages()
{
    int32 size, ret;
    uint32 code = 0;
    char *buffer = NULL;

    // read messages as long as there are messages to be read
    // or as long as there is not a source plugin assigned ( this will block )
    while ( pp_port_has_message( mMsgPort ) >= GE_NO_ERROR || mSrcPlugin == NULL )
    {
        if ( ( size = pp_port_buffer_size( mMsgPort ) ) >= GE_NO_ERROR )
        {
            // allocate space for the message
            char *p = ( char * )realloc( buffer, size );
            if ( p == NULL )
            {
                printf( "realloc() failure! %d : %s\n", __LINE__, __FILE__ );
                break;
            }
        
            buffer = p;
            // retrieve the message
            ret = pp_read_port( mMsgPort, &code, ( void * )buffer, size );

            // do something with the message
            if ( ret >= GE_NO_ERROR )
            {
                IpcMessage message( code );
                ret = message.Unflatten( buffer );
                if ( ret >= GE_NO_ERROR )
                {
                    handleMessage( &message );
                }
            }

            // check for quit
            if ( mQuitting )
            {
                break;
            }
        }
        else
        {
            break;
        }
    }

    // free the message buffer
    free( buffer );
}
//-------------------------------------------------------------
void
AppSourceThread::handleMessage( IpcMessage *msg )
{
    switch ( msg->what )
    {
        case GE_QUIT:
        {
            mQuitting = true;
            break;
        }
        case GE_ADD_FILTER:
        {
            // message contains:
            // - "filter" : BAS_POINTER_TYPE : a pointer to the new filter
            // - "successor" : BAS_POINTER_TYPE : a pointer to the filter that will run
            //      after this filter. ( may be NULL )
            ge_filter_plugin *filter = NULL, *successor = NULL;

            // fish the pointers out of the message
            msg->FindPointer( "filter", ( void const ** )&filter );
            msg->FindPointer( "successor", ( void const ** )&successor );

            // add the filter
            addFilter( filter, successor );

            break;
        }
        case GE_REMOVE_FILTER:
        {
            // message contains:
            // - "filter" : BAS_POINTER_TYPE : a pointer to the plugin instance
            // - "reply" : BAS_INT32_TYPE : a reply port number
            // - "destroy" : BAS_BOOL_TYPE : a flag indicating whether another process
            //      is waiting for a reply so that the plugin instance can be destroyed
            ipc_port_t replyPort;
            ge_filter_plugin *filter;
            bool destroy;

            // grab the data
            msg->FindPointer( "filter", ( void const ** )&filter );
            msg->FindInt32( "reply", ( ipc_port_t * )&replyPort );
            msg->FindBool( "destroy", &destroy );

            // remove the filter
            removeFilter( filter );

            // maybe this filter is going away for good?
            if ( destroy )
            {
                // change the message to a GE_DESTROY_FILTER message
                msg->what = GE_DESTROY_FILTER;
                msg->RemoveName( "reply" );
                // send the reply
                msg->SendMessage( replyPort );
            }

            break;
        }
        case GE_ADD_SOURCE:
        {
            // message contains:
            // - "source" : BAS_POINTER_TYPE : a pointer to the plugin instance
            // - "reply" : BAS_INT32_TYPE : the reply port number
            ge_source_plugin *source;
            ipc_port_t replyPort;
            
            // grab the data
            msg->FindPointer( "source", ( void const ** )&source );
            msg->FindInt32( "reply", ( ipc_port_t * )&replyPort );

            // add the source
            mSrcPlugin = source;

            // tell the receiver how this source can be reached
            msg->ReplaceInt32( "reply", mMsgPort );
            // send the reply
            msg->SendMessage( replyPort );
            
            break;
        }
        case GE_REMOVE_SOURCE:
        {
            // message contains:
            // - "source" : BAS_POINTER_TYPE : a pointer to the plugin instance
            // - "reply" : BAS_INT32_TYPE : the reply port number
            // - "destroy" : BAS_BOOL_TYPE : a flag indicating whether another process
            //      is waiting for a reply so that the plugin instance can be destroyed
            ge_source_plugin *source;
            ipc_port_t replyPort;
            bool destroy;
            
            // grab the data
            msg->FindPointer( "source", ( void const ** )&source );
            msg->FindInt32( "reply", ( ipc_port_t * )&replyPort );
            msg->FindBool( "destroy", &destroy );

            // remove the source
            if ( mSrcPlugin == source )
            {
                // point to nothing
                mSrcPlugin = NULL;

                // only send a destroy message if necessary
                if ( destroy )
                {
                    // change the message to a GE_DESTROY_SOURCE message
                    msg->what = GE_DESTROY_SOURCE;
                    msg->RemoveName( "reply" );
                    // send the reply
                    msg->SendMessage( replyPort );
                }
            }
            
            break;
        }
        case GE_CONNECT_OUTPUT:
        {
            // message contains:
            // - "output" : BAS_POINTER_TYPE : a pointer to the plugin instance
            ge_output_plugin *output;

            msg->FindPointer( "output", ( void const ** )&output );

            // connect the output
            connectOutput( output );
            
            break;
        }
        case GE_DISCONNECT_OUTPUT:
        {
            // message contains:
            // - "output" : BAS_POINTER_TYPE : a pointer to the plugin instance
            // - "reply" : BAS_INT32_TYPE : the reply port number
            // - "destroy" : BAS_BOOL_TYPE : a flag indicating whether another process
            //      is waiting for a reply so that the plugin instance can be destroyed
            ge_output_plugin *output;
            ipc_port_t replyPort;
            bool destroy;

            msg->FindPointer( "output", ( void const ** )&output );
            msg->FindInt32( "reply", &replyPort );
            msg->FindBool( "destroy", &destroy );

            // connect the output
            disconnectOutput( output );

            if ( destroy )
            {
                // change the message to a GE_DESTROY_OUTPUT message
                msg->what = GE_DESTROY_OUTPUT;
                msg->RemoveName( "reply" );
                // send the reply
                msg->SendMessage( replyPort );
            }
            
            break;
        }
        default:
        {
            printf( "unhandled message in AppSourceThread::handleMessage!"
                            " ( what = %d )\n", msg->what );
            break;
        }
    }
}
//-------------------------------------------------------------
void
AppSourceThread::addFilter( ge_filter_plugin *filt, ge_filter_plugin *before )
{
    // if 'before' is NULL, the filter goes at the end of the list
    if ( before == NULL )
    {
        mFiltList.push_back( filt );
    }
    else
    {
        // find plugin pointed to by 'before'
        for ( list<ge_filter_plugin *>::iterator i = mFiltList.begin();
                    i != mFiltList.end(); i++ )
        {
            if ( *i == before )
            {
                // insert 'filt' before 'before'
                mFiltList.insert( i, filt );
                break;
            }
        }
    }
}
//-------------------------------------------------------------
void
AppSourceThread::removeFilter( ge_filter_plugin *filt )
{
    // find the filter pointed to by 'filt'
    for ( list<ge_filter_plugin *>::iterator i = mFiltList.begin();
                i != mFiltList.end(); i++ )
    {
        // and remove it from the list
        if ( *i == filt )
        {
            mFiltList.erase( i );
            break;
        }
    }
}
//-------------------------------------------------------------
void
AppSourceThread::connectOutput( ge_output_plugin *output )
{
    // make sure there isn't already an output connected
    assert( mOutPlugin == NULL );

    // connect to the output
    mOutPlugin = output;
}
//-------------------------------------------------------------
void
AppSourceThread::disconnectOutput( ge_output_plugin *output )
{
    // make sure the correct output is being disconnected
    assert( output == mOutPlugin );

    // disconnect the output
    mOutPlugin = NULL;
}
//-------------------------------------------------------------
void *
AppSourceThread::start_loop( void *data )
{
    return ( ( AppSourceThread * )data )->main_loop();
}
//-------------------------------------------------------------
void *
AppSourceThread::main_loop()
{
    while ( true )
    {
        unsigned int tmp=0;
        bool surfacechange = false;

        // get a plugin
        //receive_data( &sender, ( void * )&plugin, sizeof( plugin ) );
        // receive any messages sent to the source
        // the messages contain: source plugins, filters, quit requests
        receiveMessages();

        // quit if we need to
        if ( mQuitting )
        {
            break;
        }
        
        // don't go any further if there isn't a source plugin ( there should be )
        if ( mSrcPlugin != NULL )
        {
            // grab a frame
            mSrcPlugin->getFrame( mSrcPlugin, mSurfacePool[mWorkSurface] );

            // did the size or format change?
            if ( *mSurfacePool[mWorkSurface] != *mSurfacePool[mScratchSurface] )
            {
                // oh no! update the scratch surface to match the work surface
                mSurfacePool[mScratchSurface]->modifyAttributes( 
                                mSurfacePool[mWorkSurface]->colorFormat(),
                                mSurfacePool[mWorkSurface]->width(),
                                mSurfacePool[mWorkSurface]->height() );

                // set a flag so that the output surface gets updated later
                surfacechange = true;
            }

            // run through the filters
            for ( list<ge_filter_plugin *>::iterator filt = mFiltList.begin();
                        filt != mFiltList.end(); filt++ )
            {
                ( *filt )->filter( *filt, mSurfacePool[mWorkSurface],
                                                    mSurfacePool[mScratchSurface] );
    
                // swap work and scratch surfaces
                tmp = mWorkSurface;
                mWorkSurface = mScratchSurface;
                mScratchSurface = tmp;
            }
    
            // swap work and output surfaces
            mSurfacePool[mOutputSurface]->lock();
            mSurfacePool[mWorkSurface]->lock();
            tmp = mOutputSurface;
            mOutputSurface = mWorkSurface;
            mWorkSurface = tmp;
            mSurfacePool[mOutputSurface]->unlock();
            mSurfacePool[mWorkSurface]->unlock();

            // ( maybe ) output the product
            if ( mOutPlugin )
            {
                mOutPlugin->output( mOutPlugin, mSurfacePool[mOutputSurface] );
            }

            // ok, what was the output surface is now the work surface
            // check to see if any adjustments need to be made
            if ( surfacechange )
            {
                mSurfacePool[mWorkSurface]->modifyAttributes( 
                                                mSurfacePool[mOutputSurface]->colorFormat(),
                                                mSurfacePool[mOutputSurface]->width(),
                                                mSurfacePool[mOutputSurface]->height() );
            }
        }
    }
    
    return NULL;
}
//-------------------------------------------------------------
